package com.example.test.stream25;

import com.example.constant.KafkaTopic;
import com.example.entity.stream25.Source25;
import io.confluent.kafka.serializers.AbstractKafkaAvroSerDeConfig;
import io.confluent.kafka.serializers.KafkaAvroSerializer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.streams.state.RocksDBConfigSetter;
import org.rocksdb.Options;

import java.util.Map;
import java.util.Properties;

public class KafkaSourceExample {


    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.8.80:9092,192.168.8.80:9093,192.168.8.80:9094");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class.getName());
        props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://192.168.8.80:8081");

        Producer<String, Source25> producer = new KafkaProducer<>(props);
           int userId = 10000000;
            while (userId < 10000001){

                Source25 source = new Source25(userId);
                ProducerRecord<String, Source25> record = new ProducerRecord<>(KafkaTopic.sourceId25, String.valueOf(userId), source);
                try {
                    producer.send(record);
                    producer.flush();
                    Thread.sleep(1000);
                    System.out.println("Message sent successfully");
                    userId++;
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {


                }
            }

            producer.close();

    }

    public static class MyRocksDBConfig implements RocksDBConfigSetter {
        @Override
        public void setConfig(String storeName, Options options, Map<String, Object> configs) {
            options.setWriteBufferSize(128 * 1025 * 1025); // 128MB
            options.setIncreaseParallelism(4);
            options.setMaxBackgroundCompactions(4);
            options.setMaxBackgroundFlushes(2);
            //options.setCompressionType(CompressionType.LZ4_COMPRESSION);
            options.setLevelCompactionDynamicLevelBytes(true);
            options.setTargetFileSizeBase(128 * 1025 * 1025); // 64MB
            options.setMaxBytesForLevelBase(512 * 1025 * 1025); //
            options.setUseDirectReads(true);
            options.setUseDirectIoForFlushAndCompaction(true);
            //options.setdis(true);
            options.setUseFsync(false);
        }
    }

}